package net.lab1024.sa.base.module.support.traffic.service;

import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson2.JSON;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import net.lab1024.sa.base.module.support.traffic.config.MqttConfig;
import net.lab1024.sa.base.module.support.traffic.domain.dto.MqttCommandDTO;
import net.lab1024.sa.base.module.support.traffic.domain.dto.MqttHeartbeatDTO;
import net.lab1024.sa.base.module.support.traffic.domain.dto.MqttHeartbeatResponseDTO;
import net.lab1024.sa.base.module.support.traffic.domain.dto.MqttResponseDTO;

/**
 * MQTT客户端服务
 */
@Slf4j
@Service
public class MqttClientService {

    @Resource
    private MqttConfig mqttConfig;

    @Resource
    private TrafficMqttMessageHandler messageHandler;

    @Resource
    private TrafficHistoryService trafficHistoryService;

    private MqttClient mqttClient;
    private final ConcurrentHashMap<String, MqttResponseDTO> responseMap = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, MqttHeartbeatDTO> heartbeatMap = new ConcurrentHashMap<>();

    /**
     * 初始化MQTT客户端
     */
    @PostConstruct
    public void init() {
        try {
            String clientId = mqttConfig.getClientIdPrefix() + UUID.randomUUID().toString().replace("-", "");
            mqttClient = new MqttClient(mqttConfig.getBrokerUrl(), clientId, new MemoryPersistence());

            MqttConnectOptions options = new MqttConnectOptions();
            options.setUserName(mqttConfig.getUsername());
            options.setPassword(mqttConfig.getPassword().toCharArray());
            options.setConnectionTimeout(mqttConfig.getConnectionTimeout());
            options.setKeepAliveInterval(mqttConfig.getKeepAliveInterval());
            options.setAutomaticReconnect(mqttConfig.isAutomaticReconnect());
            options.setCleanSession(mqttConfig.isCleanSession());

            mqttClient.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    log.error("MQTT连接丢失", cause);
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    handleMessage(topic, message);
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    log.debug("MQTT消息发送完成: {}", token.getMessageId());
                }
            });

            mqttClient.connect(options);
            log.info("MQTT客户端连接成功: {}", mqttConfig.getBrokerUrl());

            // 订阅所有设备的心跳和响应主题
            subscribeTopics();

        } catch (MqttException e) {
            log.error("MQTT客户端初始化失败", e);
            throw new RuntimeException("MQTT客户端初始化失败", e);
        }
    }

    /**
     * 订阅主题
     */
    private void subscribeTopics() {
        try {
            // 订阅所有设备的心跳主题（使用通配符）
            // String heartbeatTopic = mqttConfig.getTopicPrefix() + "/lane/lane_2/heartbeat";
            String heartbeatTopic = "traffic_control/intersection/1201100/lane/lane_2/heartbeat";
            mqttClient.subscribe(heartbeatTopic, 1);
            log.info("订阅心跳主题: {}", heartbeatTopic);

            // 订阅所有设备的响应主题（使用通配符）
            // String responseTopic = mqttConfig.getTopicPrefix() + "/lane/lane_2/response";
            String responseTopic = "traffic_control/intersection/1201100/lane/lane_2/response";
            mqttClient.subscribe(responseTopic, 1);
            log.info("订阅响应主题: {}", responseTopic);

            // 订阅设置心跳间隔响应主题
            // String setHeartbeatResponseTopic = mqttConfig.getTopicPrefix() + "/lane/lane_2/setheartbeat/response";
            String setHeartbeatResponseTopic = "traffic_control/intersection/1201100/lane/lane_2/setheartbeat/response";
            mqttClient.subscribe(setHeartbeatResponseTopic, 1);
            log.info("订阅设置心跳间隔响应主题: {}", setHeartbeatResponseTopic);

            // 订阅设置闪烁时间响应主题
            // String setFlashtimeResponseTopic = mqttConfig.getTopicPrefix() + "/lane/lane_2/setflashtime/response";
            String setFlashtimeResponseTopic = "traffic_control/intersection/1201100/lane/lane_2/setflashtime/response";
            mqttClient.subscribe(setFlashtimeResponseTopic, 1);
            log.info("订阅设置闪烁时间响应主题: {}", setFlashtimeResponseTopic);

        } catch (MqttException e) {
            log.error("订阅主题失败", e);
        }
    }

    /**
     * 处理接收到的消息
     */
    private void handleMessage(String topic, MqttMessage message) {
        try {
            String payload = new String(message.getPayload());
            log.debug("收到MQTT消息 - Topic: {}, Payload: {}", topic, payload);

            if (topic.contains("/heartbeat") && !topic.contains("/response")) {
                // 处理心跳消息
                MqttHeartbeatDTO heartbeatDTO = JSON.parseObject(payload, MqttHeartbeatDTO.class);
                heartbeatMap.put(heartbeatDTO.getRequestId(), heartbeatDTO);
                messageHandler.handleHeartbeat(topic, heartbeatDTO);
                // 发送心跳响应
                sendHeartbeatResponse(topic, heartbeatDTO.getRequestId());
            } else if (topic.contains("/response") || topic.contains("/setheartbeat/response") || topic.contains("/setflashtime/response")) {
                // 处理响应消息
                MqttResponseDTO responseDTO = JSON.parseObject(payload, MqttResponseDTO.class);
                responseMap.put(responseDTO.getRequestId(), responseDTO);
                messageHandler.handleResponse(topic, responseDTO);
            }

        } catch (Exception e) {
            log.error("处理MQTT消息失败 - Topic: {}", topic, e);
        }
    }

    /**
     * 发送心跳响应
     */
    private void sendHeartbeatResponse(String heartbeatTopic, String requestId) {
        try {
            String responseTopic = heartbeatTopic + "/response";
            MqttHeartbeatResponseDTO response = new MqttHeartbeatResponseDTO();
            response.setRequestId(requestId);
            response.setStatus("SUCCESS");
            response.setMessage("成功");
            response.setTimestamp(System.currentTimeMillis());

            String payload = JSON.toJSONString(response);
            MqttMessage message = new MqttMessage(payload.getBytes());
            message.setQos(1);
            message.setRetained(false);

            mqttClient.publish(responseTopic, message);
            log.info("发送心跳响应 - Topic: {}, RequestId: {}", responseTopic, requestId);
            trafficHistoryService.saveHeartbeatResponse(responseTopic, response);

        } catch (Exception e) {
            log.error("发送心跳响应失败", e);
        }
    }

    /**
     * 发布命令消息
     */
    public MqttResponseDTO publishCommand(String deviceRoad, Integer deviceId, MqttCommandDTO commandDTO) throws MqttException {
        String topic = buildCommandTopic(deviceRoad, deviceId, commandDTO.getCommand());
        String payload = JSON.toJSONString(commandDTO);
        MqttMessage message = new MqttMessage(payload.getBytes());
        message.setQos(1);
        message.setRetained(false);

        mqttClient.publish(topic, message);
        log.info("发布MQTT命令 - Topic: {}, RequestId: {}, Command: {}", topic, commandDTO.getRequestId(), commandDTO.getCommand());

        // 等待响应（最多等待10秒）
        return waitForResponse(commandDTO.getRequestId(), 10, TimeUnit.SECONDS);
    }

    /**
     * 等待响应
     */
    private MqttResponseDTO waitForResponse(String requestId, long timeout, TimeUnit unit) {
        long endTime = System.currentTimeMillis() + unit.toMillis(timeout);
        while (System.currentTimeMillis() < endTime) {
            MqttResponseDTO response = responseMap.remove(requestId);
            if (response != null) {
                return response;
            }
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
        // 超时返回
        MqttResponseDTO timeoutResponse = new MqttResponseDTO();
        timeoutResponse.setRequestId(requestId);
        timeoutResponse.setStatus("TIMEOUT");
        timeoutResponse.setMessage("等待响应超时");
        timeoutResponse.setTimestamp(System.currentTimeMillis());
        return timeoutResponse;
    }

    /**
     * 构建命令主题
     */
    private String buildCommandTopic(String deviceRoad, Integer deviceId, String command) {
        if ("SET_LANE_MODE".equals(command)) {
            return mqttConfig.getTopicPrefix() + deviceRoad + "/lane/lane_" + deviceId + "/command";
        } else if ("SET_INTERVAL".equals(command)) {
            return mqttConfig.getTopicPrefix() + deviceRoad + "/lane/lane_" + deviceId + "/setheartbeat";
        } else if ("SET_FLASHTIME".equals(command)) {
            return mqttConfig.getTopicPrefix() + deviceRoad + "/lane/lane_" + deviceId + "/setflashtime";
        }
        throw new IllegalArgumentException("未知的命令类型: " + command);
    }

    /**
     * 生成请求ID
     */
    public String generateRequestId() {
        return UUID.randomUUID().toString();
    }

    /**
     * 关闭MQTT客户端
     */
    @PreDestroy
    public void destroy() {
        if (mqttClient != null && mqttClient.isConnected()) {
            try {
                mqttClient.disconnect();
                mqttClient.close();
                log.info("MQTT客户端已关闭");
            } catch (MqttException e) {
                log.error("关闭MQTT客户端失败", e);
            }
        }
    }
}

